package com.ksyun.campus.metaserver.biz.services;

import cn.hutool.core.util.ObjectUtil;
import com.alibaba.fastjson2.JSON;
import com.ksyun.campus.metaserver.config.MasterManager;
import com.ksyun.campus.metaserver.config.RedisSwitchManager;
import com.ksyun.campus.metaserver.entity.server.DataServer;
import com.ksyun.campus.metaserver.entity.server.MetaServer;
import com.ksyun.campus.metaserver.mq.consumer.SyncMetaDataConsumerManager;
import com.ksyun.campus.metaserver.util.NetworkUtils;
import com.ksyun.campus.metaserver.util.ZooKeeperUtil;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.nio.charset.StandardCharsets;
import java.time.LocalDateTime;
import java.util.List;

@Slf4j
@Component
@RequiredArgsConstructor
public class RegistService implements ApplicationRunner {
    @Value("${server.port}")
    private int serverPort;
    private String serverPath;
    private final ZooKeeper zooKeeper;
    private final ZooKeeperUtil zooKeeperUtil;
    private final MasterManager masterManager;
    private final SyncMetaDataConsumerManager syncMetaDataConsumerManager;
    private final RedisSwitchManager redisSwitchManager;

    @Override
    public void run(ApplicationArguments args) throws Exception {
        registToCenter();
        System.out.println("获取dslist");
        List<DataServer> dslist = zooKeeperUtil.getDataServerlist(zooKeeperUtil.ZK_DATA_SERVER_ROOT_PATH);
        System.out.println(dslist);
        System.out.println("获取mslist");
        List<MetaServer> mslist = zooKeeperUtil.getMetaServerlist(zooKeeperUtil.ZK_META_SERVER_ROOT_PATH);
        System.out.println(mslist);
    }

    /**
     * 将本实例信息注册至zk中心，包含信息 ip、port
     */
    public void registToCenter() {
        try {
            ensurePathExists(zooKeeperUtil.ZK_META_SERVER_ROOT_PATH);  // 检查并创建父节点
            String localIp = NetworkUtils.getLocalIp();
            MetaServer metaServer = MetaServer.builder()
                    .ip(localIp)
                    .port(serverPort)
                    .build();
            // todo 可能需要考虑加锁
            {
                // 检查zk中当前是否有主节点 没有的话自己就是主
                MetaServer master = zooKeeperUtil.getMaster();
                if (ObjectUtil.isNull(master)) {
                    metaServer.setIsMaster(true);
                    masterManager.setIsMaster(true);
                    // 主节点关闭消费者
                    syncMetaDataConsumerManager.stopConsumer();
                    // 主节点使用redis数据库1号
                    redisSwitchManager.switchToMaster();
                } else {
                    // 当前有主
                    metaServer.setIsMaster(false);
                    masterManager.setIsMaster(false);
                    // 从节点启动消费者
                    syncMetaDataConsumerManager.startConsumer();
                    // 从节点使用redis数据库2号
                    redisSwitchManager.switchToSlave();
                }
            }

            // 更新zk
            String serverInfo = JSON.toJSONString(metaServer);
            serverPath = zooKeeper.create(zooKeeperUtil.ZK_META_SERVER_ROOT_PATH + "/server-",
                    serverInfo.getBytes(StandardCharsets.UTF_8),
                    ZooDefs.Ids.OPEN_ACL_UNSAFE,
                    CreateMode.EPHEMERAL_SEQUENTIAL);

            log.info("MetaServer 注册成功 [serverPath]: {} [time]: {} ", serverPath, LocalDateTime.now());

        } catch (Exception e) {
            log.error("注册到Zookeeper失败", e);

        }
    }

    /**
     * 定期上报 MetaServer 状态信息
     */
    // todo 上传100M文件时可能会断开zookeeper，尝试是否是需要schedule线程池解决
    @Scheduled(initialDelay = 30000, fixedRate = 30000) // 每30秒上报一次
    public void reportStatus() {
        try {
            // 获取本机实际IP
            String localIp = NetworkUtils.getLocalIp();
            // 更新 Zookeeper 上的节点数据
            MetaServer metaServer = MetaServer.builder()
                    .ip(localIp)
                    .port(serverPort)
                    .build();

            // todo 可能需要考虑加锁
            {
                // 检查zk中自己是不是master
                MetaServer curMetaServer = zooKeeperUtil.getCurMetaServer(serverPort);
                // 如果是空 需要自己成为主节点
                if (curMetaServer == null) {
                    masterManager.setIsMaster(true);
                }
                // 是master 继续当master
                if (curMetaServer!=null && curMetaServer.getIsMaster()) {
                    metaServer.setIsMaster(true);
                    masterManager.setIsMaster(true);
                    // 主节点关闭消费者
                    syncMetaDataConsumerManager.stopConsumer();
                    // 主节点使用redis数据库1号
                    redisSwitchManager.switchToMaster();
                } else {
                    // 但如果全局开关是主，就升级为主
                    if (masterManager.getIsMaster()) {
                        metaServer.setIsMaster(true);
                        // 主节点关闭消费者
                        syncMetaDataConsumerManager.stopConsumer();
                        // 主节点使用redis数据库1号
                        redisSwitchManager.switchToMaster();
                    } else {
                        metaServer.setIsMaster(false);
                        masterManager.setIsMaster(false);
                        // 从节点启动消费者
                        syncMetaDataConsumerManager.startConsumer();
                        // 从节点使用redis数据库2号
                        redisSwitchManager.switchToSlave();
                    }
                }
            }
            // 更新zk
            String updatedInfo = JSON.toJSONString(metaServer);
            zooKeeper.setData(serverPath, updatedInfo.getBytes(StandardCharsets.UTF_8), -1);
            log.info("MetaServer 状态已更新: " + updatedInfo);
        } catch (Exception e) {
            log.error("状态更新失败", e);
        }
    }

    // 检查路径是否存在，不存在则创建
    private void ensurePathExists(String path) throws KeeperException, InterruptedException {
        String[] pathElements = path.split("/");
        StringBuilder currentPath = new StringBuilder();
        for (String element : pathElements) {
            if (!element.isEmpty()) {
                currentPath.append("/").append(element);
                Stat stat = zooKeeper.exists(currentPath.toString(), false);
                if (stat == null) {
                    zooKeeper.create(currentPath.toString(), null, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
                }
            }
        }
    }
}
